package org.kth.dks.dks_comm;

import java.util.LinkedList;

import org.apache.log4j.Logger;

/**
 * <p>Title: DKS</p>
 * <p>Description: DKS Middleware</p>
 * <p>Copyright: Copyright (c) 2004</p>
 * <p>Company: KTH-IMIT</p>
 * @author Ali Ghodsi (aligh@kth.se)
 * @version 1.0
 */


public class ThreadPool {
	private Logger log = Logger.getLogger(ThreadPool.class);
	private final static int POOLSIZE_DEFAULT = 5;
	private int poolSize;
	private ThreadWorker workers[];
	private LinkedList jobs = new LinkedList();
	private static final int MAXJOBS = 100000;
    
    private static ThreadPool instance = null;
	
	private ThreadPool() {
		this(POOLSIZE_DEFAULT);
	}
	
	private ThreadPool(int poolSize) {
		this.poolSize = poolSize;
		workers = new ThreadWorker[poolSize];
		for (int i=0; i<poolSize; i++) {
			workers[i] = new ThreadWorker(i);
			workers[i].start();
		}
	}
	
    public static ThreadPool getInstance() {
        if (instance == null) {
            instance = new ThreadPool();
        }
        return instance;
    }
    
	public void addJob(Runnable job) {
		synchronized (jobs) {
			while (jobs.size()==MAXJOBS) {
				try {
					log.error( "Hit upper limit of number of jobs in thread pool ("+MAXJOBS+")");
					jobs.wait();
				} catch (InterruptedException ex) {
					ex.printStackTrace();
				}
			}
			jobs.addLast(job);
			
			if (jobs.size()==1)
				jobs.notify();
		}
	}
	
	public Runnable remJob() {
		Runnable job; 
		synchronized (jobs) {
			while (jobs.isEmpty()) {
				try {
					jobs.wait();
				} catch(InterruptedException ex) {
					ex.printStackTrace();
				}
			}
			job = (Runnable)jobs.removeFirst();
			if (jobs.size()==MAXJOBS-1)
				jobs.notify();
		}
		return job;
	}
	
	public void end() {
		for (int x=0; x<poolSize; x++) {
			workers[x].finish(true);
		}
	}
	
	private class ThreadWorker extends Thread {
		private boolean finish = false;
		
		public ThreadWorker(int i) {
			setName("WorkerThread-" + i);
		}
		
		public void run() {
			while (!finish()) {
				try {
					Runnable job = remJob();
//					log.debug("Serving job bef:"+job.getClass().toString());
					job.run();
//					log.debug("Serving job aft:"+job.getClass().toString());
				} catch (RuntimeException ex) {
					ex.printStackTrace();
				}
			}
		}
		
		public synchronized boolean finish() {
			return finish;
		}
		
		public synchronized void finish(boolean finish) {
			this.finish=finish;
		}
	}
	
}